package com.handydata.antifraud.channel;


import com.alibaba.fastjson.JSON;
import com.handydata.antifraud.bo.orderbaseinfo.OrderBaseInfoReqBO;
import com.handydata.antifraud.bo.orderbaseinfo.OrderBaseInfoResBO;
import com.handydata.antifraud.channel.outbound.DataPusher;
import com.handydata.antifraud.common.constant.AlarmInfoCode;
import com.handydata.antifraud.common.exception.OrderBaseInfoExCode;
import com.handydata.antifraud.rule.RuleEngineScheduler;
import com.handydata.antifraud.service.AlarmInfoService;
import com.handydata.antifraud.service.DataCallLogsService;
import com.handydata.antifraud.service.DecisionExecRecordService;
import com.handydata.antifraud.service.OrderBaseInfoService;
import com.handydata.antifraud.to.RiskOutputTO;
import com.handydata.antifraud.util.IPUtil;
import com.handydata.antifraud.vo.riskres.RiskOutputVO;
import com.handydata.common.convert.CommonBeanConvertor;
import com.handydata.common.util.RedisUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Objects;
import java.util.concurrent.Future;

@Slf4j
@Component
public class RabbitReceiver {

    @Autowired
    private OrderBaseInfoService orderBaseInfoService;

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private RuleEngineScheduler ruleEngineScheduler;

    @Autowired
    private DataPusher dataPusher;

    @Autowired
    private AlarmInfoService alarmInfoService;
    @Autowired
    private DecisionExecRecordService decisionExecRecordService;
    @Autowired
    private DataCallLogsService dataCallLogsService;
    @Value("${risk.shbank.sleepTime}")
    private int sleepTime;

//    @Value("${spring.rabbitmq.queue}")
//    private String port;
    
    @Value("${server.port}")
    private String port;

    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "queue-1",
                durable="true"),
        exchange = @Exchange(value = "exchange-2",
                durable="true",
                type= "topic",
                ignoreDeclarationExceptions = "true"),
        key = "springboot.abc"
    )
    )

    @RabbitHandler
    public void onMessage(Message<?> message, Channel channel) throws Exception {
        long start =System.currentTimeMillis();
        log.info("接收到消息开始执行--------------------------------------getChannelNumber {} consumer hash {}",channel.getChannelNumber(),this.hashCode());//重试机制，业务完成，消息失败也会重搞
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        OrderBaseInfoReqBO orderBaseInfoReqBO=null;
        String redisKey=null;
        try{
            try{
                orderBaseInfoReqBO = (OrderBaseInfoReqBO)JSON.parseObject((String)message.getPayload(),OrderBaseInfoReqBO.class);
                log.info("订单号ordeId{},开始计算处理订单",orderBaseInfoReqBO.getOrderId());
                if(Objects.isNull(orderBaseInfoReqBO)||orderBaseInfoReqBO.getOrderId()==null
                        ||orderBaseInfoReqBO.getOrderId().length()<=0) {
                    log.error("---------------消息队列取出订单信息为空 {}---------------",orderBaseInfoReqBO);
                    //throw new NullPointerException("订单信息不能为空");
                    channel.basicAck(deliveryTag, false);
                    return;
                }
            }catch (Exception ex){
                channel.basicAck(deliveryTag, false);
                log.error("转换数据出错:{}，应该是垃圾消息"+ex.getMessage(),message.getPayload());
                return;
            }
            redisKey="order_" + orderBaseInfoReqBO.getOrderId();
            log.info("接收到消息开始执行--------------------------------------{}",orderBaseInfoReqBO.getOrderId());
            Integer flag=(Integer)redisUtil.get(redisKey);
            if((flag!=null)&&(flag==2||flag==4||flag==5)) {
                //正在执行的不再执行了 0 接受入库，1正在执行，2执行成功，3执行失败，4发送成功，5发送失败
                log.info("接收到消息开始执行：该订单已执行过，不再执行--------------------------------------{}",orderBaseInfoReqBO.getOrderId());
                channel.basicAck(deliveryTag, false);
                return;
            }
            orderBaseInfoReqBO.setStartTime(new Date());
            orderBaseInfoReqBO.setProcessState("1");
            orderBaseInfoReqBO.setCacl_ip(IPUtil.getIpAddress() + ":" + port);
            log.info("更新订单号orderId{},开始时间", orderBaseInfoReqBO.getOrderId());
            updateTDb(orderBaseInfoReqBO);
            log.info("更新订单号orderId{},结束时间", orderBaseInfoReqBO.getOrderId());
            log.info("---------------更新redis订单状态  {}---------------",orderBaseInfoReqBO);
            setnx(redisKey, 1);
            Future<RiskOutputTO> future=ruleEngineScheduler.runRuleEngine(orderBaseInfoReqBO);
			RiskOutputTO riskOutputTO = future.get();
            Long dual=System.currentTimeMillis()-start;
            long needSleep=sleepTime-dual;
            try {
                if(needSleep>0){
                    Thread.sleep(needSleep);
                    log.info("---------------------------配合上行接收订单处理结果，睡眠{} 执行时间{}-------------------------------------",needSleep,dual);
                }
            }catch (Exception ex){
                channel.basicAck(deliveryTag, false);
                log.error("执行订单睡眠出问题",ex);
            }
            if(riskOutputTO!=null){
                log.info("接收到消息已执行成功--------------------------------------{}",orderBaseInfoReqBO.getOrderId());
                dataPusher.push(riskOutputTO);
                RiskOutputVO riskOutputVO = CommonBeanConvertor.convert(riskOutputTO, RiskOutputVO.class);
                OrderBaseInfoReqBO reqBO = CommonBeanConvertor.convert(riskOutputVO, OrderBaseInfoReqBO.class);
                reqBO.setSendMsg(riskOutputVO.getJSONString());
                reqBO.setEndTime(new Date());
                reqBO.setProcessState("2");
                updateTDb(reqBO);
                setnx(redisKey, 2);
            }else {
                log.error("执行订单出错 {}",orderBaseInfoReqBO);
                alarmInfoService.sendWatchAbNormal(AlarmInfoCode.ALERT_ITEM_ORDER.getCode(),orderBaseInfoReqBO.getOrderId(),null,
                        AlarmInfoCode.ALARM_TYPE_ERR.getCode(),"订单执行异常",null);
                riskOutputTO=new RiskOutputTO();
                riskOutputTO.setOrderId(orderBaseInfoReqBO.getOrderId());
                RiskOutputVO riskOutputVO = CommonBeanConvertor.convert(riskOutputTO, RiskOutputVO.class);
                orderBaseInfoReqBO.setEndTime(new Date());
                orderBaseInfoReqBO.setProcessState("3");
                orderBaseInfoReqBO.setSendMsg(riskOutputVO.getJSONString());
                updateTDb(orderBaseInfoReqBO);
                setnx(redisKey, 3);
                dataPusher.push(riskOutputTO);//上行居然没有失败标志
            }
            channel.basicAck(deliveryTag, false);
        }catch (Exception ex){
            channel.basicAck(deliveryTag, false);
            if(orderBaseInfoReqBO!=null){
                RiskOutputTO riskOutputTO=new RiskOutputTO();
                riskOutputTO.setOrderId(orderBaseInfoReqBO.getOrderId());
                orderBaseInfoReqBO.setEndTime(new Date());
                orderBaseInfoReqBO.setProcessState("3");
                RiskOutputVO riskOutputVO = CommonBeanConvertor.convert(riskOutputTO, RiskOutputVO.class);
                orderBaseInfoReqBO.setSendMsg(riskOutputVO.getJSONString());
                updateTDb(orderBaseInfoReqBO);
                dataPusher.push(riskOutputTO);//上行居然没有失败标志
                setnx(redisKey, 3);
                alarmInfoService.sendWatchAbNormal(AlarmInfoCode.ALERT_ITEM_ORDER.getCode(),
                        orderBaseInfoReqBO.getOrderId(),null,AlarmInfoCode.ALARM_TYPE_ERR.getCode(),"计算规则出错",ex);
            }
            log.error("----------------------处理消息时出错:{}--------------------------------------",message,ex);
        }


    }
    private void updateTDb(OrderBaseInfoReqBO reqBO){
        try{
            orderBaseInfoService.modifyOrderBaseInfoBy(reqBO);
        }catch (Exception ex){
            log.error("更新订单状态失败{}",reqBO.getOrderId(),ex);
        }
    }
    @SuppressWarnings("unused")
	private void saveDataToDb(OrderBaseInfoReqBO orderBaseInfoReqBO) {
        OrderBaseInfoResBO orderBaseInfoResBO = orderBaseInfoService.createOrderBaseInfo(orderBaseInfoReqBO);
        if(!orderBaseInfoResBO.getRespCode().equals(OrderBaseInfoExCode.SUCCESS)) {
            //异常处理
            log.warn(orderBaseInfoResBO.getRespCode(), orderBaseInfoResBO.getRespMsg());
            log.info("Save reqBO {} not success", orderBaseInfoReqBO);
        }
    }

    private void setnx(String key, Object value) {
    	if (redisUtil.hasKey(key)) {
    		redisUtil.set(key,value, 86400);
		} else {
			redisUtil.set(key,value,86400);
		}
    	
//       return redisUtil.setIfAbsent(key,value);
    }

}